package com.ngkj.zmfptback.business.aop;

import com.alibaba.fastjson.JSONArray;
import com.ngkj.zmfptCommon.common.bean.RpcRequestMessage;
import com.ngkj.zmfptback.business.bean.RentalRequestEntiry;
import com.ngkj.zmfptback.business.bean.RentalResponseEntity;
import com.ngkj.zmfptback.business.mapper.read.RentalReadMapper;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.After;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * @author ChenHN
 * @since 2022/1/17 creation
 **/
@Component
@Aspect
@Slf4j
public class RentalEsAspect {

    @Resource
    private RentalReadMapper rentalReadMapper;

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    private static String topicName = "zmfpt-es";

    @After("execution(public * com.ngkj.zmfptback.business.service.write.RentalWriteService.agreeRentals(..)) || execution(public * com.ngkj.zmfptback.business.service.write.RentalWriteService.refusedRentals(..)))")
    public void agreeRentals(JoinPoint joinPoint) {
        Object[] args = joinPoint.getArgs();
        List list1 = (List) Collections.singletonList(args[0]).get(0);
        ArrayList<Long> list = new ArrayList<>(list1);
        List<RentalResponseEntity> rentalResponseEntities = rentalReadMapper.getRentalByListId(list);
        sendToEs(rentalResponseEntities);
    }

    /**
     * 同意全部
     * @param pjp
     * @throws Throwable
     */
    @Around("execution(public * com.ngkj.zmfptback.business.service.write.RentalWriteService.agreeAllRental(..))")
    public void agreeAllRental(ProceedingJoinPoint pjp) throws Throwable {
        List<Long> list = rentalReadMapper.getAllNoProcessedRentalListId();
        pjp.proceed();
        List<RentalResponseEntity> rentalResponseEntities = rentalReadMapper.getRentalByListId((ArrayList<Long>) list);
        sendToEs(rentalResponseEntities);
    }

    /**
     * 新增切面
     * @param joinPoint
     */
    @After("execution(public * com.ngkj.zmfptback.business.service.write.RentalWriteService.addRentalInfo(..))")
    public void addRentalInfo(JoinPoint joinPoint){
        Object[] args = joinPoint.getArgs();
        RentalRequestEntiry rentalRequestEntiry = (RentalRequestEntiry) args[0];
        Long czfxxlsh = rentalRequestEntiry.getCzfxxlsh();
        List<Long> list = new ArrayList<>();
        list.add(czfxxlsh);
        List<RentalResponseEntity> rentalByListId = rentalReadMapper.getRentalByListId((ArrayList<Long>) list);
        sendToEs(rentalByListId);
    }

    /**
     * 取消发布
     * @param joinPoint
     */
    @After("execution(public * com.ngkj.zmfptback.business.service.write.RentalWriteService.cancelRental(..))")
    public void cancelRental(JoinPoint joinPoint){
        Object[] args = joinPoint.getArgs();
        Long aLong = (Long) args[0];
        List<Long> list = new ArrayList<>();
        list.add(aLong);
        List<RentalResponseEntity> rentalByListId = rentalReadMapper.getRentalByListId((ArrayList<Long>) list);
        sendToEs(rentalByListId);
    }


    /**
     * 将参数构造，放入kafka中，利用反射操作
     * @param rentalResponseEntities
     */
    private void sendToEs(List<RentalResponseEntity> rentalResponseEntities) {
        String jsonString = JSONArray.toJSONString(rentalResponseEntities);
        RpcRequestMessage rpcRequestMessage = new RpcRequestMessage(
                "com.ngkj.zmfptes.service.impl.RentalInfoServiceImpl",
                "insertAndUpdateRental",
                Boolean.class,
                new Class[]{String.class},
                new Object[]{jsonString}
        );
        kafkaTemplate.send(topicName, rpcRequestMessage).addCallback((success) -> {
                    //消息发送到的主题
                    String topic = success.getRecordMetadata().topic();
                    //消息发送的分区
                    int partition = success.getRecordMetadata().partition();
                    log.info("发送消息成功：{} - {} ", topic, partition);
                }, (failure) -> {
                    log.warn("发送消息失败：{}", failure.getMessage());
                }
        );
    }

}
